package com.econ.powercloud.ems.observer;

import com.econ.powercloud.ems.common.enums.MeasureType;
import com.econ.powercloud.ems.common.pojo.EmsKafkaYcDataDTO;
import com.econ.powercloud.ems.constant.SysConstant;
import com.econ.powercloud.ems.common.interf.AnalogDataObserver;
import com.econ.powercloud.ems.websocket.WebsocketObserverManager;
import com.google.common.collect.Maps;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.user.SimpSession;
import org.springframework.messaging.simp.user.SimpUser;
import org.springframework.messaging.simp.user.SimpUserRegistry;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 遥测实时数据推送
 */
@Component
public class AnalogDataWebsocketProcessor implements AnalogDataObserver {

    @Autowired
    private WebsocketObserverManager websocketObserverManager;
    @Autowired
    private SimpUserRegistry simpUserRegistry;
    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Override
    public void doInvoke(List<EmsKafkaYcDataDTO> dataList) {
        //获取sessionId对应点号订阅数据,key:sessionId,value:订阅点号数据
        Map<String,Set<String>> sessionIdToPidMap = Maps.newHashMap();
        //获取sessionId对应的simpUserName,key:sessionId,value:simpUserName
        Map<String,String> sessionIdToSimpUserNameMap = Maps.newHashMap();
        simpUserRegistry.getUsers().stream().forEach(simpUser ->{
            simpUser.getSessions().stream().forEach(session ->{
                Set<String> pidSet = websocketObserverManager.getPidSetBySessionId(session.getId());
                sessionIdToPidMap.put(session.getId(),pidSet);
                sessionIdToSimpUserNameMap.put(session.getId(),simpUser.getName());
            });
        });
        dataList.stream().forEach(data ->{
            data.setSvgKey(new String[]{data.getDeviceCode(),data.getPid().replace(data.getDeviceCode() + SysConstant.SPOT,"")});
            sessionIdToPidMap.forEach((sessionId,pidSet)->{
                //二、判断是否需要推送
                if (CollectionUtils.isEmpty(pidSet)) {
                    return;
                }
                //三、判断是否为订阅的数据
                if(pidSet.contains(data.getPid())) {
                    SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
                    headerAccessor.setSessionId(sessionId);
                    headerAccessor.setLeaveMutable(true);
                    messagingTemplate.convertAndSendToUser(sessionIdToSimpUserNameMap.get(sessionId), "/topic/device/analog", data, headerAccessor.getMessageHeaders());
                }
            });
        });
    }

}

